package com.bucks;

import com.bucks.model.Coffee;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
import org.springframework.data.redis.core.ReactiveHashOperations;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.jdbc.core.JdbcTemplate;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.CountDownLatch;

@Slf4j
@SpringBootApplication
public class SpringBucksReactorApplication implements ApplicationRunner {

    private static final String KEY = "COFFEE_MENU";

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Autowired
    private ReactiveStringRedisTemplate reactiveStringRedisTemplate;

    public static void main(String[] args) {
        SpringApplication.run(SpringBucksReactorApplication.class, args);
    }

    @Bean
    ReactiveStringRedisTemplate reactiveStringRedisTemplate(ReactiveRedisConnectionFactory factory) {
        return new ReactiveStringRedisTemplate(factory);
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        ReactiveHashOperations hashOperations = reactiveStringRedisTemplate.opsForHash();
        CountDownLatch countDownLatch = new CountDownLatch(1);

        List<Coffee> list = jdbcTemplate.query(
                "select * from t_coffee", (resultSet, rowNum) ->
                        Coffee.builder()
                                .id(resultSet.getLong("id"))
                                .name(resultSet.getString("name"))
                                .price(resultSet.getLong("price"))
                                .build()
        );


        Flux.fromIterable(list)
                .publishOn(Schedulers.single())
                .doOnComplete(() -> log.info("list ok "))
                .flatMap(coffee -> {
                    log.info("存放数据 {} , {}", coffee.getName(), coffee.getPrice());
                    return hashOperations.put(KEY, coffee.getName(), coffee.getPrice().toString());
                })
                .doOnComplete(() -> log.info("set ok"))
                .concatWith(reactiveStringRedisTemplate.expire(KEY, Duration.ofMinutes(1)))
                .doOnComplete(() -> log.info("expire ok"))
                .onErrorResume(e -> {
                    log.error("exception {}", e);
                    return Mono.just(false);
                })
                .subscribe(coffee -> log.info("接收到结果: {}", coffee.toString()),
                        e -> log.error("Error {}", e.toString()),
                        countDownLatch::countDown
                );
        log.info("Waiting");
        countDownLatch.await();

    }


}
